package de.uni_stuttgart.ipvs.ids.replication;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import de.uni_stuttgart.ipvs.ids.communication.ReadRequestMessage;
import de.uni_stuttgart.ipvs.ids.communication.ReleaseReadLock;
import de.uni_stuttgart.ipvs.ids.communication.ReleaseWriteLock;
import de.uni_stuttgart.ipvs.ids.communication.RequestReadVote;
import de.uni_stuttgart.ipvs.ids.communication.RequestWriteVote;
import de.uni_stuttgart.ipvs.ids.communication.ValueResponseMessage;
import de.uni_stuttgart.ipvs.ids.communication.Vote;
import de.uni_stuttgart.ipvs.ids.communication.WriteRequestMessage;

/**
 * Class which is handling a Replica of a {@link VersionedValue}.<br>
 * Processes client requests and replies to them.
 * 
 * @param <T>
 *            The type for the {@link VersionedValue} this replica handles.
 * @author Team 3/4 Elkhan Dadashov, Ahmed Labib, Dhaval A Shah, Patrick Stortz
 */
public class Replica<T> extends Thread {

	/** Defines whether a Replica is not locked, read locker or write locked. */
	public enum LockType {
		/** Replica is not locked. */
		UNLOCKED,
		/** Replica is read locked. */
		READLOCK,
		/** Replica is write locked. */
		WRITELOCK
	};

	/** The identifier of this replica. */
	private int id;
	/** The DatagramSocket to be used for communication with the clients. */
	private DatagramSocket socket = null;
	/** The {@link VersionedValue} of this replica. */
	private VersionedValue<T> value;
	/** The current {@link LockType} of this replica. */
	private LockType lock;
	/**
	 * Defines the possibility to handle a received DatagramPacket.
	 * 1-availability is the possibility to miss a received DatagramPacket.
	 */
	private double availability;
	/** Indicates weather a received DatagramPacket has to be missed or not. */
	private boolean missPacket = false;
	/**
	 * The {@link Random} to be used to realize the availability of this
	 * replica.
	 */
	final Random random = new Random(System.currentTimeMillis());
	/**
	 * This address holds the address of the client holding the lock. This
	 * variable should be set to NULL every time the lock is set to UNLOCKED.
	 */
	private SocketAddress lockHolder;

	/**
	 * Instantiates a Replica.
	 * 
	 * @param id
	 *            The identifier of this replica.
	 * @param listenPort
	 *            The socket port to listen to.
	 * @param availability
	 *            The availability of this replica. E.g. 0.7 means the replica
	 *            will reply to 70% of the received packets.
	 * @param initialValue
	 *            The initial value for the {@link VersionedValue} of this
	 *            replica.
	 * @throws SocketException
	 *             Thrown if the socket cannot be created/bound.
	 */
	public Replica(int id, int listenPort, double availability, T initialValue)
			throws SocketException {
		super("Replica-" + id + " @ " + listenPort);
		this.id = id;
		SocketAddress socketAddress = new InetSocketAddress("127.0.0.1",
				listenPort);
		this.socket = new DatagramSocket(socketAddress);
		this.availability = availability;
		this.value = new VersionedValue<T>(0, initialValue);
		this.lock = LockType.UNLOCKED;
	}

	/** @return The identifier of this replica. */
	public int getID() {
		return id;
	}

	/** @return The {@link SocketAddress} used by this replica. */
	public SocketAddress getSocketAddress() {
		return socket.getLocalSocketAddress();
	}

	@Override
	public void run() {

		// System.out.println(this.getName() + " | Replica running...");// DEBUG

		// create a thread pool for handling the received packets
		ExecutorService executor = Executors.newCachedThreadPool();
		while (true) {
			// create a buffer with 1KB
			byte[] buffer = new byte[1024];
			// create the packet to receive
			final DatagramPacket receivedPacket = new DatagramPacket(buffer,
					buffer.length);
			try {
				// wait for a packet
				socket.receive(receivedPacket);

				// handle the received packet in a new thread
				executor.execute(new Runnable() {
					@Override
					public void run() {
						// System.out.println(Thread.currentThread());
						handleDatagrampacket(receivedPacket);
					}
				});
			} catch (IOException e) {
				System.err.println(this.getName() + " | " + "  Exception: "
						+ e.getMessage());
				e.printStackTrace();
			}
		}
	}

	/**
	 * Handles a DatagramPacket received from a client.
	 * 
	 * @param packet
	 *            The DatagramPacket to handle.
	 */
	private synchronized void handleDatagrampacket(DatagramPacket packet) {

		// realize the availability, check if we are unlocked (node failure only
		// if unlocked)
		missPacket = false;
		if (this.lock == LockType.UNLOCKED) {
			// generate a (uniform distributed) double in the range 0.0 - 1.0,
			// and check if lower 1-availability (realizes 1-availability miss)
			if (random.nextDouble() < (1 - this.availability)) {
				// miss this packet
				missPacket = true;
				// System.out.println(this.getName()
				// + " | Replica missing a packet from "
				// + packet.getSocketAddress());// DEBUG
			}
		}

		// check if we have to handle the received packet
		if (missPacket == false) {
			try {
				// deserialize the received object
				Object object = getObjectFromMessage(packet);

				// check the instance of the received object and handle it
				// appropriately
				if (object instanceof RequestWriteVote) {
					handleWriteVoteRequest((RequestWriteVote) object,
							packet.getSocketAddress());
				} else if (object instanceof RequestReadVote) {
					handleReadVoteRequest((RequestReadVote) object,
							packet.getSocketAddress());
				} else if (object instanceof ReleaseWriteLock) {
					handleWriteLockRelease((ReleaseWriteLock) object,
							packet.getSocketAddress());
				} else if (object instanceof ReleaseReadLock) {
					handleReadLockRelease((ReleaseReadLock) object,
							packet.getSocketAddress());
				} else if (object instanceof WriteRequestMessage<?>) {
					handleWriteRequest((WriteRequestMessage<?>) object,
							packet.getSocketAddress());
				} else if (object instanceof ReadRequestMessage) {
					handleReadRequest((ReadRequestMessage) object,
							packet.getSocketAddress());
				} else {
					throw new Exception(packet.getSocketAddress()
							+ " sent a unknown request message (Class: "
							+ object.getClass() + ")!");
				}
			} catch (ClassNotFoundException e) {
				System.err.println(this.getName() + " | " + "  Exception: "
						+ e.getMessage());
				e.printStackTrace();
			} catch (IOException e) {
				System.err.println(this.getName() + " | " + "  Exception: "
						+ e.getMessage());
				e.printStackTrace();
			} catch (Exception e) {
				System.err.println(this.getName() + " | " + "  Exception: "
						+ e.getMessage());
				e.printStackTrace();
			}
		}

	}

	/**
	 * Handles a RequestWriteVote received from a client.<br>
	 * Sets the write lock and replies with an YES Vote to the requesting client
	 * if the replica is not locked.<br>
	 * Replies with an NO Vote to the requesting client if the replica is read
	 * or write locked.<br>
	 * 
	 * @param writeVoteRequest
	 *            the RequestWriteVote to handle.
	 * @param address
	 *            the address from which the RequestWriteVote has been received.
	 * @throws IOException
	 *             Thrown if the sending of the YES/NO Vote fails.
	 */
	private void handleWriteVoteRequest(RequestWriteVote writeVoteRequest,
			SocketAddress address) throws IOException {

		// check if no one holds the exclusive read or write lock
		if (this.lock == LockType.UNLOCKED) {
			// exclusively write lock the replica
			this.lock = LockType.WRITELOCK;
			this.lockHolder = address;
			// send a YES vote the requesting client
			sendObjectOverUDP(address,
					new Vote(Vote.State.YES, this.value.getVersion()));
			// System.out.println(this.getName()
			// + " | Replica got write locked by " + address);// DEBUG
		} else {
			// someone holds a exclusive read or write lock
			// send a NO vote the requesting client
			sendObjectOverUDP(address, new Vote(Vote.State.NO, -1));
			System.out.println(this.getName()
					+ " | Replica declined write lock from " + address);// DEBUG
		}
	}

	/**
	 * Handles a RequestReadVote received from a client.<br>
	 * Sets the read lock and replies with an YES Vote to the requesting client
	 * if the replica is not locked or only read locked.<br>
	 * Replies with an NO Vote to the requesting client if the replica is write
	 * locked.<br>
	 * 
	 * @param readVoteRequest
	 *            the RequestReadVote to handle.
	 * @param address
	 *            the address from which the RequestReadVote has been received.
	 * @throws IOException
	 *             Thrown if the sending of the YES/NO Vote fails.
	 */
	private void handleReadVoteRequest(RequestReadVote readVoteRequest,
			SocketAddress address) throws IOException {

		// check if no one holds the exclusive read or write lock
		if (this.lock == LockType.UNLOCKED) {
			// exclusively read lock the replica
			this.lock = LockType.READLOCK;
			this.lockHolder = address;
			// send a YES vote the requesting client
			sendObjectOverUDP(address,
					new Vote(Vote.State.YES, this.value.getVersion()));
			// System.out.println(this.getName()
			// + " | Replica got read locked by " + address);// DEBUG
		} else {
			// someone holds a exclusive read or write lock
			// send a NO vote the requesting client
			sendObjectOverUDP(address, new Vote(Vote.State.NO, -1));
			System.out.println(this.getName()
					+ " | Replica declined read lock from " + address);// DEBUG
		}
	}

	/**
	 * Handles a ReleaseWriteLock received from a client.<br>
	 * Releases the write lock and replies with an ACK to the requesting client
	 * if request is valid.<br>
	 * Replies with an NACK to the requesting client if request is invalid.
	 * 
	 * @param writeLockRelease
	 *            the ReleaseWriteLock to handle.
	 * @param address
	 *            the address from which the ReleaseWriteLock has been received.
	 * @throws Exception
	 *             Thrown if there is no write lock holder or the requesting
	 *             client is not the write lock holder.
	 * @throws IOException
	 *             Thrown if the sending of the ACK/NACK fails.
	 */
	private void handleWriteLockRelease(ReleaseWriteLock writeLockRelease,
			SocketAddress address) throws Exception, IOException {

		// check if someone holds the exclusive write lock
		if (this.lock == LockType.WRITELOCK) {
			// check if the requesting client holds the exclusive write lock
			if (this.lockHolder.equals(address)) {
				// release the exclusive write lock
				this.lock = LockType.UNLOCKED;
				this.lockHolder = null;
				// send an ACK to the requesting client
				sendAcknowledgment(address, true);
				// System.out.println(this.getName()
				// + " | Replica got write unlocked by " + address);// DEBUG
			} else {
				// logic error
				// the requesting client does not hold the exclusive write lock
				// sent an NACK to the requesting client
				sendAcknowledgment(address, false);
				throw new Exception(
						address
								+ " wants to release the write lock, but does not own the write lock!"
								+ " Lock Releaser: " + address
								+ ", Lock Holder: " + this.lockHolder);
			}
		} else {
			// logic error
			// nobody holds the exclusive write lock
			// sent an NACK to the requesting client
			sendAcknowledgment(address, false);
			throw new Exception(address
					+ " wants to release a write lock which is not existing!");
		}
	}

	/**
	 * Handles a ReleaseReadLock received from a client.<br>
	 * Releases the read lock and replies with an ACK to the requesting client
	 * if request is valid.<br>
	 * Replies with an NACK to the requesting client if request is invalid.
	 * 
	 * @param readLockRelease
	 *            the ReleaseReadLock to handle.
	 * @param address
	 *            the address from which the ReleaseReadLock has been received.
	 * @throws Exception
	 *             Thrown if there is no read lock holder or the requesting
	 *             client is not the read lock holder.
	 * @throws IOException
	 *             Thrown if the sending of the ACK/NACK fails
	 */
	private void handleReadLockRelease(ReleaseReadLock readLockRelease,
			SocketAddress address) throws Exception, IOException {

		// check if someone holds the exclusive read lock
		if (this.lock == LockType.READLOCK) {
			// check if the requesting client holds the exclusive read lock
			if (this.lockHolder.equals(address)) {
				// release the exclusive read lock
				this.lock = LockType.UNLOCKED;
				this.lockHolder = null;
				// send an ACK to the requesting client
				sendAcknowledgment(address, true);
				// System.out.println(this.getName()
				// + " | Replica got read unlocked by " + address);// DEBUG
			} else {
				// logic error
				// the requesting client does not hold the exclusive read lock
				// sent an NACK to the requesting client
				sendAcknowledgment(address, false);
				throw new Exception(
						address
								+ " wants to release a read lock, but does not own a read lock!"
								+ " Lock Releaser: " + address
								+ ", Lock Holder: " + this.lockHolder);
			}
		} else {
			// logic error
			// nobody holds the exclusive read lock
			// sent an NACK to the requesting client
			sendAcknowledgment(address, false);
			throw new Exception(address
					+ " wants to release a read lock which is not existing!");
		}
	}

	/**
	 * Handles a WriteRequestMessage received from a client.<br>
	 * Updates the local value and sends an ACK to the requesting client if
	 * request is valid.<br>
	 * Furthermore it sends an NACK to the requesting client if the request is
	 * invalid.
	 * 
	 * @param writeRequest
	 *            the WriteRequestMessage to handle.
	 * @param address
	 *            the address from which the WriteRequestMessage has been
	 *            received.
	 * @throws Exception
	 *             Thrown if there is no lock holder or the requesting client is
	 *             not the lock holder.
	 * @throws IOException
	 *             Thrown if the sending of the ACK/NACK fails
	 */
	@SuppressWarnings("unchecked")
	// generic types are lost during compilation, we cannot check against it
	private void handleWriteRequest(WriteRequestMessage<?> writeRequest,
			SocketAddress address) throws Exception {

		// check if someone holds the exclusive write lock
		if (this.lock == LockType.WRITELOCK) {
			// check if the requesting client is the exclusive lock holder
			if (this.lockHolder.equals(address)) {
				// check if the version of the value to write is higher than our
				// local one
				if (this.value.getVersion() < writeRequest.getVersion()) {
					// update the local value to the received one
					this.value = ((VersionedValue<T>) writeRequest);
					// send an ACK to the client
					sendAcknowledgment(address, true);
					// System.out.println(this.getName()
					// + " | Replica value got updated to "
					// + this.value.getValue() + " (version "
					// + this.value.getVersion() + ") by " + address);// DEBUG
				}
			} else {
				// logic error
				// the requesting client does not hold the exclusive write lock
				// sent an NACK to the requesting client
				sendAcknowledgment(address, false);
				throw new Exception(
						address
								+ " wants to write the value, but is not holding a write lock!"
								+ " Reader: " + address + ", Lock Holder: "
								+ this.lockHolder);
			}

		} else {
			// logic error
			// nobody holds the exclusive write lock
			// sent an NACK to the requesting client
			sendAcknowledgment(address, false);
			throw new Exception(address
					+ " wants to write the value while it is not write locked!");
		}
	}

	/**
	 * Handles a ReadRequestMessage received from a client.<br>
	 * Replies with a ValueResponseMessage to the requesting client if request
	 * is valid.
	 * 
	 * @param readRequest
	 *            the ReadRequestMessage to handle.
	 * @param address
	 *            the address from which the ReadRequestMessage has been
	 *            received.
	 * @throws Exception
	 *             Thrown if there is no lock holder or the requesting client is
	 *             not the lock holder.
	 * @throws IOException
	 *             Thrown if the sending of the value fails
	 */
	private void handleReadRequest(ReadRequestMessage readRequest,
			SocketAddress address) throws Exception, IOException {

		// check if someone holds the exclusive read or write lock
		if (this.lock == LockType.READLOCK || this.lock == LockType.WRITELOCK) {
			// check if the requesting client is the exclusive lock holder
			if (this.lockHolder.equals(address)) {
				// reply the value to the requesting client
				ValueResponseMessage<T> valueMessage = new ValueResponseMessage<T>(
						this.value.getValue());
				sendObjectOverUDP(address, valueMessage);
				// System.out.println(this.getName()
				// + " | Replica value got read by " + address);// DEBUG
			} else {
				// logic error
				// the requesting client does not hold the exclusive read or
				// write lock
				throw new Exception(
						address
								+ " wants to read the value, but is not holding a read or write lock!"
								+ " Reader: " + address + ", Lock Holder: "
								+ this.lockHolder);
			}
		} else {
			// logic error
			// nobody holds the exclusive read or write lock
			throw new Exception(
					address
							+ " wants to read the value while it is not read or write locked!");
		}
	}

	/**
	 * Sends an acknowledge or non-acknowledge to a client.
	 * 
	 * @param address
	 *            The address of the client.
	 * @param acknowledge
	 *            True if an acknowledge has to be sent, false if an
	 *            non-acknowledge has to be sent.
	 * @throws IOException
	 *             Thrown if sending the acknowledge/non-acknowledge fails.
	 */
	private void sendAcknowledgment(SocketAddress address, boolean acknowledge)
			throws IOException {
		if (acknowledge) {
			sendObjectOverUDP(address, new Vote(Vote.State.YES, -1));
		} else {
			sendObjectOverUDP(address, new Vote(Vote.State.NO, -1));
		}
	}

	/**
	 * Serializes an object and sends it over the UDP socket.
	 * 
	 * @param address
	 *            The address the UDP packet has to be sent to.
	 * @param object
	 *            The object to serialize and to put in the packet data
	 * @throws IOException
	 *             Thrown if there are errors during writing the stream.
	 */
	private void sendObjectOverUDP(SocketAddress address, Object object)
			throws IOException {

		// create the byte and object stream
		ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
		ObjectOutputStream objectOutputStream = new ObjectOutputStream(
				byteArrayOutputStream);
		// write the object to the object stream
		objectOutputStream.writeObject(object);
		objectOutputStream.flush();
		// get the bytes of the object and create the packet
		byte[] buffer = byteArrayOutputStream.toByteArray();
		DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
		packet.setSocketAddress(address);
		// send the packet
		socket.send(packet);
		// close the object stream
		objectOutputStream.close();
	}

	/**
	 * Deserializes an object from a DatagramPacket.
	 * 
	 * @param packet
	 *            The DatagramPacket with the serialized object in its data.
	 * @return The deserialized object
	 * @throws IOException
	 *             Thrown if there are errors during reading the stream.
	 * @throws ClassNotFoundException
	 *             Thrown if there is no object in the packet data.
	 */
	private static Object getObjectFromMessage(DatagramPacket packet)
			throws IOException, ClassNotFoundException {
		// get the byte stream
		ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(
				packet.getData());
		// get the object stream
		ObjectInputStream objectInputStream = new ObjectInputStream(
				byteArrayInputStream);
		// read the object from the stream
		Object object = objectInputStream.readObject();
		// close the steam
		objectInputStream.close();
		return object;
	}
}
